package cn.doitedu.day01

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.DriverManager


object T08_QueryCategoryName {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("QueryCategoryName")
      .setMaster("local[5]") //如果提交到集群中运行，setMaster必须注释掉

    val sc = new SparkContext(conf)

    val lines: RDD[String] = sc.textFile("data/order.txt")

    //在客户端窗口的MySQL链接对象(不能再Driver创建链接，因为链接对象无法序列化，导致Task出现错误)
    //Class.forName("com.mysql.cj.jdbc.Driver")
    //val connection = DriverManager.getConnection("jdbc:mysql://node-1.51doit.cn:3306/doit?characterEncoding=utf-8", "root", "123456")

    val resRdd: RDD[(String, Long, Double, String)] = lines.map(line => {
      //将创建链接的代码放到函数的里面，就不会出现错误了，因为链接对象是在Executor中创建
      //但是存在的问题就是，每来一条数据，创建一个连接对象，效率低
      Class.forName("com.mysql.cj.jdbc.Driver")
      val connection = DriverManager.getConnection("jdbc:mysql://node-1.51doit.cn:3306/doit?characterEncoding=utf-8", "root", "123456")

      //函数在Executor
      val fields = line.split(",")
      val oid = fields(0)
      val cid = fields(1).toLong
      val money = fields(2).toDouble
      val preparedStatement = connection.prepareStatement("select name from tb_category where id = ?")
      preparedStatement.setLong(1, cid)
      val resultSet = preparedStatement.executeQuery()
      var name: String = null
      if (resultSet.next()) {
        name = resultSet.getString(1)
      }
      resultSet.close()
      preparedStatement.close()
      connection.close()
      (oid, cid, money, name)
    })



    resRdd.saveAsTextFile("out/out07")


  }

}
